package com.niit.statis.result

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

// 定义Product样例类
case class Product( productId: Int, name: String, imageUrl: String, categories: String, tags: String )
//定义统计出来历史热门商品样例类
case class RateMoreProduct(productId: Int,total: Long)
//定义统计出来历史热门商品样例类
case class RateMoreRecentlyProduct(productId: Int,total: Long,monthOfYear:Long)
// 定义mongo配置类
case class MongoConfig(uri:String, db:String)

object StatisticRecommenderResult {

  val RATE_MORE_PRODUCTS = "RateMoreProducts" // 定义历史热门商品表，通过计算商品的评分次数来推测
  val RATE_MORE_RECENTLY_PRODUCTS = "RateMoreRecentlyProducts" // 定义最近热门商品表（以月为单位进行统计）
  val MONGODB_PRODUCT_COLLECTION = "Product" // 商品表
  val MONGODB_PRODUCT_RATING_MORE_COLLECTION = "ProductAndRateMore" // 历史热门商品评分次数表
  val MONGODB_PRODUCT_RECONTLY_RATING_COLLECTION = "ProductAndRecentlyRateMore" // 最近热门商品评分次数表

  def main(args: Array[String]): Unit = {
    val config = Map( // 将mongodb配置信息用Map来定义，由于我的mongodb开启了权限认证，故需要指定用户名与密码
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://fooadmin:123456@127.0.0.1:27017/bigdata",
      "mongo.db" -> "bigdata"
    )
    // 创建SparkConf配置
    val sparkConf = new SparkConf().setAppName("StatisticRecommenderResult").setMaster(config("spark.cores"))
    // 创建SparkSession
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    // 初始化MongoConfig,声明为implicit
    implicit  val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    // 加入隐式转换
    import spark.implicits._

    // spark对象读取mongodb中的Product表的数据，将其转换成DF
    val productDF = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_PRODUCT_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Product]
      .toDF()

    val rateMoreProductDF = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", RATE_MORE_PRODUCTS)
      .format("com.mongodb.spark.sql")
      .load()
      .as[RateMoreProduct]
      .toDF()

    val recentlyRateMoreProductDF = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", RATE_MORE_RECENTLY_PRODUCTS)
      .format("com.mongodb.spark.sql")
      .load()
      .as[RateMoreRecentlyProduct]
      .toDF()


    // 创建临时表（在SparkSession存活期间有效）
    productDF.createOrReplaceTempView("products")
    rateMoreProductDF.createOrReplaceTempView("rateMoreProducts")
    recentlyRateMoreProductDF.createOrReplaceTempView("recentlyRateProducts")

    // TODO: 不同的统计推荐结果
    // 1. 历史热门商品统计
    statisticsHistory(spark)
    // 2. 近期热门商品
    recentlyHotProductsStatistics(spark)
    // 停止
    spark.stop()
  }

  // 1. 历史热门商品统计，按照商品的评分个数进行简单统计,并将统计结果写入到mongodb中
  // productId, name,imageUrl,tags
  def statisticsHistory(spark: SparkSession)(implicit mongoConfig: MongoConfig): Unit ={
    val historyDF = spark.sql("select p.productId,p.name,p.imageUrl,p.categories,p.tags, r.total from products p,rateMoreProducts r where p.productId = r.productId order by r.total desc")
    insertIntoMongodb(historyDF,MONGODB_PRODUCT_RATING_MORE_COLLECTION)
  }

  // 2. 近期热门商品，把时间戳转换成yyyyMM格式进行评分个数统计，最终得到productId, count, yearmonth
  def recentlyHotProductsStatistics(spark: SparkSession)(implicit mongoConfig: MongoConfig): Unit ={
    val recentHotProductsDF = spark.sql("select p.productId,p.name,p.imageUrl,p.categories,p.tags,r.monthOfYear,r.total from products p,recentlyRateProducts r where p.productId = r.productId order by r.monthOfYear desc,r.total desc")
    // 2.6 调用方法写入mongo数据库表RateMoreRecentlyProducts中
    insertIntoMongodb(recentHotProductsDF,MONGODB_PRODUCT_RECONTLY_RATING_COLLECTION)
  }

  // 自定义入库函数，用于将df存入mongodb中
  def insertIntoMongodb(df: DataFrame,collection_name: String)(implicit mongoConfig: MongoConfig): Unit ={
    df.write
      .option("uri",mongoConfig.uri)
      .option("collection",collection_name)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()
  }

}
